package com.navinfo.tripanalysis.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.navinfo.tripanalysis.common.util.CommonUtils;
import com.navinfo.tripanalysis.pojo.LoadDataParam;
import com.navinfo.tripanalysis.pojo.Point0200;
import com.navinfo.tripanalysis.pojo.Point;
import com.navinfo.tripanalysis.pojo.StatusAddition;
import com.navinfo.tripanalysis.util.StatusAdditionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import static com.navinfo.tripanalysis.common.protocol.java.LCStatusType.StatusType.*;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

/**
 * 加载0200的基础服务
 * @author 沈东生
 */
public abstract class AbstractLoad0200DataServiceBigData implements Load0200DataService {
    protected static final List<String> COL_NAMES = new ArrayList<>();
    private final String SERNUMBER_NAME = "serialnumber";
    /**
     * 是否支持serialnumber字段
     */
    private boolean supportSerialNumber;

    public AbstractLoad0200DataServiceBigData() {
        COL_NAMES.addAll(Arrays.asList("gpsdate", "status", "additionalarm", "statusaddition", "mileage", "ispatch", "speed", "height", "originallat", "originallng", "direction", "electricvehicle", "standardmileage", "standardfuelcon", "tcuload", "vecuload"));
    }

    public boolean isSupportSerialNumber() {
        return supportSerialNumber;
    }

    public void setSupportSerialNumber(boolean supportSerNumber) {
        if (supportSerNumber) {
            if (!COL_NAMES.contains(SERNUMBER_NAME)) {
                COL_NAMES.add(SERNUMBER_NAME);
            }
        } else {
            if (COL_NAMES.contains(SERNUMBER_NAME)) {
                COL_NAMES.remove(SERNUMBER_NAME);
            }
        }
        this.supportSerialNumber = supportSerNumber;
    }

    /**
     * 获取过滤条件
     * @param param
     * @return
     */
    protected  String getWhereSql0200(LoadDataParam param) {
        String whereSql = "gpsdate!=0 and gpsdate is not null and terminalid!=0 and terminalid is not null ";
        List<Long> tids = param.getTids();
        if (tids != null && tids.size() > 0) {
            StringBuilder sb = new StringBuilder();
            sb.append(String.format(" and terminalid in(%s)", com.navinfo.tripanalysis.util.CommonUtils.mkString(tids, ",")));
            whereSql += sb.toString();
        }
        return whereSql;
    }

    protected JavaRDD<Point> toJavaRdd(Dataset<Row> dataset) {
        return dataset.toJavaRDD().mapPartitions((FlatMapFunction<Iterator<Row>, Point>) iterator -> new Iterator<Point>() {
                    @Override
                    public boolean hasNext() {
                        return iterator.hasNext();
                    }
                    @Override
                    public Point next() {
                        return getPoint(iterator.next());
                    }
                }
        );
    }

    private Point getPoint(Row row) {
        Point0200 point = new Point0200();
        point.setTid(row.getAs("terminalid"));
        point.setProtocol(Point.PROTOCOL_0200);
        point.setGpsTime(((long)row.getAs("gpsdate")) * 1000L);
        point.setHeight(Double.valueOf(row.getAs("height").toString()));
        point.setLatitude(Integer.valueOf(row.getAs("originallat").toString()));
        point.setLongitude(Integer.valueOf(row.getAs("originallng").toString()));
        point.setDirection(Integer.valueOf(row.getAs("direction").toString()));
        //标准速度，标准速度为0取gps速度
        double speed = Double.parseDouble(row.getAs("electricvehicle").toString()) / 100.0;
        point.setSpeedKmh(speed==0 ? Double.parseDouble(row.getAs("speed").toString()) : speed);

        //0200特有，GPS里程
        point.setGpsMileage( BigDecimal.valueOf(row.getAs("mileage")).divide(
                new BigDecimal("1000"), 1, BigDecimal.ROUND_DOWN).doubleValue());

        //0200特有，标准总里程，总油耗
        point.setStandardMileage(Float.parseFloat(row.getAs("standardmileage")));
        point.setStandardFuelCons(Float.parseFloat(row.getAs("standardfuelcon")));


        //0200特有，是否补传，acc状态
        point.setPatch(row.getAs("ispatch"));
        point.setAccStatus((int) CommonUtils.getToBinary(row.getAs("status"), 0));

        //车辆状态附加信息
        List<StatusAddition> statusAdditions = StatusAdditionUtils.parseToList(row.getAs("statusaddition"));
        for (StatusAddition statusAddition : statusAdditions) {
            long statusValue = statusAddition.getValue();
            switch(Integer.parseInt(statusAddition.getKey())) {
                case rotation_VALUE:
                    //当前转速
                    point.setRotation(new BigDecimal(statusValue).divide(new BigDecimal("100"), 0, BigDecimal.ROUND_HALF_UP).intValue());
                    break;
                case mileage_VALUE:
                    //0200特有，can里程(ecu)
                    point.setEcuMileage(statusValue / 100.0);
                    break;
                case mileageDD_VALUE:
                    point.setMeterMileage(statusValue / 100.0);//0200特有，仪表里程
                    break;
                case differentialMileage_VALUE:
                    point.setIntegralMileage(statusValue / 100.0);//0200特有，积分里程，不使用
                    break;
                case totalFuelConsumption_VALUE:
                    point.setTotalFuelCons(statusValue / 100.0);//0200特有，燃油总消耗
                    break;
                case integralFuelConsumption_VALUE:
                    point.setIntegralFuelCons( statusValue / 100.0);//0200特有，积分油耗
                    break;
                case accPedalPos_VALUE:
                    point.setAccPedalPos((int) statusValue / 100);//油门开度
                    break;
                case coolingWaterTem_VALUE:
                    point.setCoolingWaterTem((int) statusValue / 100);//0200特有，冷却水温
                    break;
                case intakeAirTem_VALUE:
                    point.setIntakeAirTem((int) statusValue / 100); //0200特有，进气温度
                    break;
                case dpfPressure_VALUE:
                    point.setIntakePressure((int) statusValue / 100); //0200特有，进气压力
                    break;
                case clutchSwitch_VALUE:
                    point.setClutchSwitch((int) statusValue / 100); //离合器开关
                    break;
                case cruiseCtrlActive_VALUE:
                    point.setCruiseControl(statusValue == 1); //0200特有，定速巡航开关
                    break;
                case oilValue_VALUE:
                    //0200特有，油液百分比
                    double per = statusValue / 100.0;
                    if (per >= 0 && per <= 100) {
                        point.setOilPercentage(per);
                    }
                    break;
                case fuelConsumptionRate_VALUE:
                    point.setFuelConsumptionRate(statusValue / 100.0 / 1000.0); //发动机燃油消耗率（毫升/小时）转为 L/H
                    break;
                case realTimeOilConsumption_VALUE:
                    point.setRealTimeOilConsumption(statusValue / 100.0);//实时油耗(发动机瞬时油耗)（KM/L）
                    break;
                case actualEnginePercentTor_VALUE:
                    point.setEngineOutputTorque((int) statusValue / 100); //实际发动机转矩百分比
                    break;
                default:
                    break;
            }
        }

        //附加报警位
        String additionAlarm = row.getAs("additionalarm");
        if (StringUtils.startsWith(additionAlarm, "[")) {
            List<Byte> dataList = JSON.parseObject(additionAlarm,  new TypeReference<List<Byte>>() {});
            if (dataList.size() >= 8) {
                //0200特有，在第8个字节 Bit1 手刹开关
                point.setHandBrakeStatus((dataList.get(7) >> 1) & 0x01);
            }
            if (dataList.size() >= 6) {
                //在第6个字节
                byte additionAlarmAt6 = dataList.get(5);
                //bit0 空挡滑行，0200特有
                point.setSlideNeutralStatus(additionAlarmAt6 & 0x01);
                //bit1 急加速，0200特有
                point.setRapidAccelerationStatus((additionAlarmAt6 >> 1) & 0x01);
                //bit2 急减速，0200特有
                point.setRapidDecelerationStatus((additionAlarmAt6 >> 2) & 0x01);
                //bit3 急转弯，0200特有
                point.setSharpTurningStatus((additionAlarmAt6 >> 3) & 0x01);
                //bit6 猛踩油门报警状态，0200特有
                point.setSlamAcceleratorStatus((additionAlarmAt6 >> 6) & 0x01);
            }
        }

        //0200特有，序列号
        if (isSupportSerialNumber()) {
            Integer number = row.getAs(SERNUMBER_NAME);
            if (number != null) {
                point.setSerialNumber(number);
            }
        }

        //0200特有,AMT、VECU载重
        try{
            point.setAmtLoad(Integer.valueOf(row.getAs("tcuload").toString()));
        }catch (NumberFormatException e){
            point.setAmtLoad(0);
        }
        try{
            point.setVecuLoad(Integer.valueOf(row.getAs("vecuload").toString()));
        }catch (NumberFormatException e){
            point.setVecuLoad(0);
        }


        return point;
    }

}
